package com.atguigu.tingshu.search.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.BaseCategoryFeignClient;
import com.atguigu.tingshu.album.client.TrackInfoFeignClient;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.kafka.KafkaMessagePojo;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.order.OrderInfo;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.model.user.UserInfo;
import com.atguigu.tingshu.order.client.OrderInfoFeignClient;
import com.atguigu.tingshu.search.dao.AlbumInfoDao;
import com.atguigu.tingshu.search.dao.SuggestIndexDao;
import com.atguigu.tingshu.search.service.ItemService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import jakarta.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.redisson.api.RBloomFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

/**
 * es数据的写操作
 */
@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class ItemServiceImpl implements ItemService {

  @Autowired
  private AlbumInfoDao albumInfoDao;

  @Resource
  private AlbumInfoFeignClient albumInfoFeignClient;

  @Resource
  private UserInfoFeignClient userInfoFeignClient;

  @Resource
  private BaseCategoryFeignClient baseCategoryFeignClient;

  @Autowired
  private ThreadPoolExecutor threadPoolExecutor;

  @Autowired
  private RBloomFilter<Object> rBloomFilter;
  /**
   * 专辑上架
   * @param albumId
   */
  @Override
  public void albumInfoFromMysqlToEs(Long albumId) {
    // 构建es索引的实体对象
    AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
    // 异步编排: 查询专辑的数据
    CompletableFuture<AlbumInfo> future1 = CompletableFuture.supplyAsync(() -> {
      // 调用feign 查询albumInfo
      AlbumInfo albumInfo = albumInfoFeignClient.getAlbumInfo(albumId);
      // 专辑数据为空直接结束方法
      if (albumInfo == null || albumInfo.getId() == null) {
        return null; // 要有返回值
      }
      //专辑id
      albumInfoIndex.setId(albumInfo.getId());
      //专辑的标题
      albumInfoIndex.setAlbumTitle(albumInfo.getAlbumTitle());
      //专辑描述
      albumInfoIndex.setAlbumIntro(albumInfo.getAlbumIntro());
      //专辑封面
      albumInfoIndex.setCoverUrl(albumInfo.getCoverUrl());
      //专辑包含声音总数
      albumInfoIndex.setIncludeTrackCount(albumInfo.getIncludeTrackCount());
      //专辑是否完结：0-否；1-完结
      albumInfoIndex.setIsFinished(albumInfo.getIsFinished().toString());
      //付费类型：免费、vip免费、付费
      albumInfoIndex.setPayType(albumInfo.getPayType());
      // 返回 albumInfo
      return albumInfo;
    }, threadPoolExecutor);
    // 执行任务2-作者名字: 查询作者的信息
    CompletableFuture<Void> future2 =
        future1.thenAcceptAsync(
            albumInfo -> {
              if (albumInfo == null) {
                return;
              }
              // 查询userInfo
              UserInfo userInfo = userInfoFeignClient.getUserInfo(albumInfo.getUserId());
              if (userInfo == null) {
                albumInfoIndex.setAnnouncerName("尚硅谷默认作者");
              } else {
                albumInfoIndex.setAnnouncerName(userInfo.getNickname());
              }
            },
            threadPoolExecutor);
    // 执行任务3-分类信息: 查询一二三级的分类的信息
    CompletableFuture<Void> future3 = future1.thenAcceptAsync(albumInfo -> {
      if (albumInfo == null) {
        return;
      }
      // 查询专辑分类视图
      BaseCategoryView baseCategoryView =
              baseCategoryFeignClient.getBaseCategoryView(albumInfo.getCategory3Id());
      albumInfoIndex.setCategory1Id(baseCategoryView.getCategory1Id());
      albumInfoIndex.setCategory2Id(baseCategoryView.getCategory2Id());
      albumInfoIndex.setCategory3Id(baseCategoryView.getCategory3Id());
    }, threadPoolExecutor);
    // 执行任务4-统计信息: 查询专辑的统计信息
    CompletableFuture<Void> future4 = future1.thenAcceptAsync(albumInfo -> {
      if (albumInfo == null) {
        return;
      }
      // 查询专辑统计数据map
      Map<String, String> albumStatMap =
              albumInfoFeignClient.getAlbumStat(albumInfo.getId());
      // 播放量
      Long play = Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_PLAY));
      albumInfoIndex.setPlayStatNum(play);
      // 订阅量
      Long sub = Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_SUBSCRIBE));
      albumInfoIndex.setSubscribeStatNum(sub);
      // 浏览量
      Long buy = Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_BROWSE));
      albumInfoIndex.setBuyStatNum(buy);
      // 评论量
      Long comment = Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_COMMENT));
      albumInfoIndex.setCommentStatNum(comment);
      // 热度值计算: 播放+1 订阅+2 购买+10 评论+5
      albumInfoIndex.setHotScore((double) (play + sub * 2 + buy * 10 + comment * 5));
    }, threadPoolExecutor);
    // 执行任务5-标签数据: 查询专辑标签数据
    CompletableFuture<Void> future5 = future1.thenAcceptAsync(albumInfo -> {
      if (albumInfo == null) {
        return;
      }
      // 查询专辑属性值列表
      List<AlbumAttributeValue> albumAttributeValueList =
              albumInfoFeignClient.getAlbumAttributeValue(albumInfo.getId());
      albumInfoIndex.setAttributeValueIndexList(
              // 遍历属性值集合 封装到AttributeValueIndexList中
              albumAttributeValueList.stream().map(albumAttributeValue -> {
                // 专辑属性值的index对象
                AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId());
                attributeValueIndex.setValueId(albumAttributeValue.getValueId());
                return attributeValueIndex;
              }).toList()
      );
    }, threadPoolExecutor);
    // 设置创建时间
    albumInfoIndex.setCreateTime(new Date());
    // 所有任务执行完成进行到下一步
    CompletableFuture.allOf(future2, future3, future4, future5).join();
    // 保存数据到es中
    albumInfoDao.save(albumInfoIndex);

    // 1107 提示词最终是为了搜索专辑 提示词从专辑中找到: 专辑的名字 专辑的简介 作者名字 -> 专辑的名字
    initSuggestInfo(albumInfoIndex.getAlbumTitle(), albumInfoIndex.getAlbumTitle());
    // 作者名字
    initSuggestInfo(albumInfoIndex.getAlbumTitle(), albumInfoIndex.getAnnouncerName());
    // // 专辑的简介
    // initSuggestInfo(albumInfoIndex.getAlbumTitle(), albumInfoIndex.getAlbumIntro());

    // 布隆过滤器的新增
    rBloomFilter.add("albumId:" + albumId);
  }

  @Autowired
  private SuggestIndexDao suggestIndexDao;
  /**
   * 提示词初始化
   * @param title
   * @return
   */
  private void initSuggestInfo(String title, String content) {
    // 初始化es的提示词实体
    SuggestIndex suggestIndex = new SuggestIndex();
    // 补全属性
    // 文档下标
    suggestIndex.setId(IdWorker.get32UUID());
    // 提示的内容
    suggestIndex.setTitle(title);
    // 关键字的汉字
    suggestIndex.setKeyword(new Completion(new String[]{content}));
    // 关键字的拼音
    suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(content)}));
    // 关键字的首字母
    suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(content)}));
    // 保存提示词
    suggestIndexDao.save(suggestIndex);
  }

  /**
   * 专辑下架
   * @param albumId
   */
  @Override
  public void removeAlbumInfoFromEs(Long albumId) {
    // 删除专辑
    albumInfoDao.deleteById(albumId);
  }

  @Autowired
  private KafkaTemplate kafkaTemplate;

  @Autowired
  private OrderInfoFeignClient orderInfoFeignClient;

  @Autowired
  private TrackInfoFeignClient trackInfoFeignClient;

  @Autowired
  private RedisTemplate redisTemplate;
  /**
   * 订单完成后 增加es中的专辑统计数据
   * @param orderNo
   */
  @Override
  public void updateAlbum(String orderNo) {
    //查询订单的数据
    OrderInfo orderInfo = orderInfoFeignClient.getOrderInfo(orderNo);
    // 订单信息为空 或 VIP订单 直接返回
    if(orderInfo == null || orderInfo.getItemType().equals("1003")){
      return;
    }
    //查询主体id(a.专辑 b.声音)--->专辑
    Long itemId = orderInfo.getOrderDetailList().get(0).getItemId();
    if(orderInfo.getItemType().equals("1002")){
      //查询声音的详情,获取专辑的id
      itemId = trackInfoFeignClient.getTrackInfo(itemId).getAlbumId();
    }
    //记录哪个专辑发生了统计的变化,数据不进入es: 日期: yyyy-mm-dd
    String date = new DateTime().toString("yyyyMMdd");
    redisTemplate.opsForSet().add("Album_Stat_Change_Set_" + date, itemId.toString());
    KafkaMessagePojo kafkaMessagePojo = new KafkaMessagePojo();
    kafkaMessagePojo.setAlbumId(itemId);
    kafkaMessagePojo.setType("buy");
    //记录专辑的购买次数用于排行榜展示: 75个key: 一级分类+(热度 播放 收藏 评论 购买)
    // kafkaTemplate.send("tingshu_stream_rank_topic", 0,"rank.buy", itemId.toString());
    kafkaTemplate.send("tingshu_stream_rank_topic", JSONObject.toJSONString(kafkaMessagePojo));
  }

  /**
   * 同步专辑的统计数据
   * @param albumId
   */
  @Override
  public void updateByTask(Long albumId) {
    //查询专辑的数据
    Optional<AlbumInfoIndex> op = albumInfoDao.findById(albumId);
    if(op.isPresent()){
      //获取专辑
      AlbumInfoIndex albumInfoIndex = op.get();
      //查询最新的统计信息
      Map<String, String> albumStatMap = albumInfoFeignClient.getAlbumStat(albumId);
      Long play = Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_PLAY));
      albumInfoIndex.setPlayStatNum(play);
      Long sub = Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_SUBSCRIBE));
      albumInfoIndex.setSubscribeStatNum(sub);
      Long buy = Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_BROWSE));
      albumInfoIndex.setBuyStatNum(buy);
      Long comment = Long.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_COMMENT));
      albumInfoIndex.setCommentStatNum(comment);
      //热度值计算: 播放+1 订阅+2 购买+10 评论+5
      albumInfoIndex.setHotScore(Double.valueOf(play + sub * 2 + buy * 10 + comment * 5));
      //覆盖
      albumInfoDao.save(albumInfoIndex);
    }
  }
}
